package edu.nepu.flink.api.state;

import edu.nepu.flink.api.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.PriorityQueue;

/**
 * @Date 2024/3/2 10:42
 * @Created by chenshuaijun
 */
public class ListStateDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> source = env.socketTextStream("hadoop102", 9999).map(new MapFunction<String, WaterSensor>() {

            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                return element.getTs() * 1000;
            }
        }));
        /**
         * TODO 使用ListState要做的需求
         *          针对每种传感器输出最高的3个水位值
         */
        source.keyBy(WaterSensor::getId).process(new KeyedProcessFunction<String, WaterSensor, String>() {
            ListState<Integer> maxThree;

            @Override
            public void open(Configuration parameters) throws Exception {
                maxThree = getRuntimeContext().getListState(new ListStateDescriptor<Integer>("max-three",Types.INT));
            }

            @Override
            public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {

                maxThree.add(value.getVc());
                Iterable<Integer> sensorVcs = maxThree.get();
                ArrayList<Integer> list = new ArrayList<>();
                for (Integer vc : sensorVcs) {
                    list.add(vc);
                }
                list.sort((o1, o2) -> o2 - o1);
                if (list.size() > 3){
                    list.remove(3);
                }
                maxThree.update(list);
                out.collect("id="+value.getId()+"的3个最高的vc为"+ Arrays.toString(list.toArray()));
            }
        }).print();

        env.execute();
    }
}
